package com.apps.sdses.flink141.demo;

import com.alibaba.fastjson2.JSONObject;
import com.google.gson.Gson;
import org.apache.flink.api.common.functions.RichMapFunction;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.types.Row;

import java.util.HashMap;
import java.util.List;
import java.util.Map;

/**
 * @author jiwei
 * @description
 * @date 2023/8/24 17:00
 */
public class FieldsToJsonMapFunction extends RichMapFunction<Row, Row> {
    Gson gson = null;


    List<FieldsToJsonDemo.Column> fieldsToJson;

    public FieldsToJsonMapFunction(List<FieldsToJsonDemo.Column> fieldsToJson) {
        this.fieldsToJson = fieldsToJson;
    }

    @Override
    public void open(Configuration parameters) throws Exception {
         gson = new Gson();
    }

    @Override
    public Row map(Row row) throws Exception {
        System.out.println("-----------------value: " + row.toString());
        Map<String, Object> jsonMap = new HashMap<>();
        for (FieldsToJsonDemo.Column column : fieldsToJson) {
            String name = column.getName();
            Object value = row.getFieldAs(name);
            System.out.println("====key: " + name + "=======value: " + value);
            jsonMap.put(name, value);
        }
//        String json = gson.toJson(jsonMap);
        String json = JSONObject.toJSONString(jsonMap);

        return Row.of(json, "aaaa");
    }
}
